package com.evoupsight.monitorpass.gateway.services;

import com.evoupsight.monitorpass.gateway.constants.InterfaceTypeEnum;
import com.evoupsight.monitorpass.gateway.dao.mapper.InterfaceMapper;
import com.evoupsight.monitorpass.gateway.dao.mapper.MonitorItemMapper;
import com.evoupsight.monitorpass.gateway.dao.mapper.RelationInterfaceServerMapper;
import com.evoupsight.monitorpass.gateway.dao.mapper.ServerMapper;
import com.evoupsight.monitorpass.gateway.dao.model.*;
import com.evoupsight.monitorpass.gateway.dto.JmxCheckDTO;
import com.evoupsight.monitorpass.gateway.queue.KafkaProducerThread;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.management.*;
import javax.management.openmbean.CompositeData;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author evoup
 */
@SuppressWarnings("Duplicates")
@Service
public class JmxPollerService {
    private static final Logger LOG = Logger.getLogger(JmxPollerService.class);
    private static final Pattern JMX_ITEM_KEY_PTN = Pattern.compile("jmx\\[\"(.*)\",(.*)\\]");
    @Autowired
    @Qualifier("jmxExecutorServiceThreadPool")
    protected ExecutorService es;
    @Autowired
    private ServerMapper serverMapper;
    @Autowired
    private MonitorItemMapper monitorItemMapper;
    @Autowired
    private InterfaceMapper interfaceMapper;
    @Autowired
    private RelationInterfaceServerMapper relationInterfaceServerMapper;
    @Autowired
    @Qualifier("new_kafka_producer_pool")
    ExecutorService kafkaProducerPool;
    @Autowired
    @Qualifier("new_kafka_producer")
    KafkaProducer<String, String> kafkaProducer;

    @Value("${kafka.topic}")
    String topic;

    @SuppressWarnings("Duplicates")
    public void poll() {
        ServerExample example = new ServerExample();
        // 排除不要监控的
        example.createCriteria().andStatusNotEqualTo(2);
        List<Server> servers = serverMapper.selectByExample(example);
        List<ItemGetTask> callableTasks = new ArrayList<>();
        servers.stream().filter(Objects::nonNull).forEach(s -> {
            // todo 去掉用户关闭的监控项
            // 找该服务器的全部监控项
            MonitorItemExample itemExample = new MonitorItemExample();
            itemExample.createCriteria().andHostIdEqualTo(s.getId()).andDataFromEqualTo(2);
            List<MonitorItem> monitorItems = monitorItemMapper.selectByExample(itemExample);
            // 每一个监控项都有机会被多个接口查询，key不能在多一个接口上相同命名，否则数值就会被覆盖
            monitorItems.stream().filter(Objects::nonNull).forEach(it -> {
                RelationInterfaceServerExample relationInterfaceServerExample = new RelationInterfaceServerExample();
                relationInterfaceServerExample.createCriteria().andServerIdEqualTo(s.getId());
                List<RelationInterfaceServer> relationInterfaceServers = relationInterfaceServerMapper.selectByExample(relationInterfaceServerExample);
                if ((relationInterfaceServers.size() > 0)) {
                    for (RelationInterfaceServer relationInterfaceServer : relationInterfaceServers) {
                        Long interfaceId = relationInterfaceServer.getInterfaceId();
                        InterfaceExample interfaceExample = new InterfaceExample();
                        interfaceExample.createCriteria().andIdEqualTo(interfaceId).andTypeEqualTo(InterfaceTypeEnum.JMX.ordinal());
                        List<Interface> interfaces = interfaceMapper.selectByExample(interfaceExample);
                        if (interfaces.size() > 0) {
                            Interface anInterface = interfaces.get(0);
                            JmxCheckDTO dto = JmxCheckDTO.builder().hostName(s.getName()).hostIp(s.getIp()).jmxAddr(anInterface.getAddr()).monitorItem(it).build();
                            callableTasks.add(new ItemGetTask(dto));
                        }
                    }
                }
            });
        });

        try {
            es.invokeAll(callableTasks);
            LOG.info("jmx poll done");
        } catch (InterruptedException e) {
            e.printStackTrace();
            LOG.error(e.getMessage(), e);
        }
    }


    /**
     * 获取jmx指标的数值
     *
     * @param hostName 主机名，需要最后发送到opentsdb
     * @param jmxAddr  jmx地址，如172.16.30.220:9211
     * @param item     监控项
     */
    @SuppressWarnings("Duplicates")
    private void monitorItemJmxPoll(String hostName, String hostIp, String jmxAddr, MonitorItem item) throws IOException {
        String jmxURL = "service:jmx:rmi:///jndi/rmi://" + jmxAddr + "/jmxrmi";
        JMXServiceURL serviceURL = new JMXServiceURL(jmxURL);
        try (JMXConnector connector = connectWithTimeout(serviceURL, 4)) {
            //tomcat jmx url
            MBeanServerConnection mbsc = connector.getMBeanServerConnection();
            //端口最好是动态取得
            //ObjectName threadObjName = new ObjectName("Catalina:type=ThreadPool,name=http-8089");

            // jmx["java.lang:type=GarbageCollector,name=ConcurrentMarkSweep",CollectionCount]
            String openTsdbKey = item.getKey();
            Matcher m = getMatcher(openTsdbKey);
            if (m == null) {
                return;
            }

            String attrValue = getUniversalAttr(m, serviceURL);
            if (NumberUtils.isCreatable(attrValue)) {
                // TODO 需要上传到opentsdb，但是这2个指标暂时不能处理，要存到hbase里
                //jmx metric VmName:Java HotSpot(TM) 64-Bit Server VM
                //jmx metric VmVersion:25.191-b12
                // TODO 邻近的数据去重
                ProducerRecord<String, String> record;
                // String s = "apps.backend.lab1.script.pid_cpu_usage 1571996734 20.0";
                StringBuffer s = new StringBuffer();
                // only in key which has : scenario, opentsdb bug! already specify : in tsd.core.tag.allow_specialchars, but exception while querying
                s.append("jmx apps.backend.").append(hostName).append(".").append(openTsdbKey.replace(":", "_").replace("\"", "'")).append(" ").append(System.currentTimeMillis() / 1000).append(" ").append(attrValue).append(" host=").append(hostName).append(" ip=").append(hostIp);
                record = new ProducerRecord<>(topic, s.toString());
                kafkaProducerPool.submit(new KafkaProducerThread(kafkaProducer, record));
            } else {
                LOG.info("这个数据暂时不能上传");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 简单和复合mbean属性都能获取
     *
     * @param m
     * @param serviceUrl
     * @return
     */
    private String getUniversalAttr(Matcher m, JMXServiceURL serviceUrl) {
        String metricValue = "";
        try (JMXConnector connector = connectWithTimeout(serviceUrl, 4)) {
            // java.lang:type=Memory
            String objectNameStr = m.group(1);
            // HeapMemoryUsage.used
            String attrName = m.group(2);
            LOG.info(objectNameStr + "," + attrName);
            //tomcat jmx url
            MBeanServerConnection mbsc = connector.getMBeanServerConnection();
            if ((attrName.contains("."))) {
                String[] dotedAttr = attrName.split("\\.");
                if (dotedAttr.length != 2) {
                    LOG.info("没有.，不是复合的JMX属性类型");
                    return metricValue;
                }
                // HeapMemoryUsage
                String attribute = dotedAttr[0];
                // used
                String compositeDataKey = dotedAttr[1];
                try {
                    Object o = mbsc.getAttribute(new ObjectName(objectNameStr), attribute);
                    CompositeData cd = (CompositeData) o;
                    metricValue = cd.get(compositeDataKey).toString();
                    LOG.info("jmx metric:" + metricValue);
                } catch (InstanceNotFoundException ignore) {
                    LOG.info("该mbean下没有该属性，忽略");
                } catch (Exception e) {
                    e.printStackTrace();
                    LOG.error(e.getMessage(), e);
                }
            } else {
                try {
                    metricValue = mbsc.getAttribute(new ObjectName(objectNameStr), attrName).toString();
                } catch (InstanceNotFoundException ignore) {
                    LOG.info("该mbean下没有该属性，忽略");
                } catch (MBeanException | AttributeNotFoundException | ReflectionException | MalformedObjectNameException e) {
                    e.printStackTrace();
                }
                LOG.info("jmx metric:" + metricValue);
            }
        } catch (InterruptedException | ExecutionException | IOException | TimeoutException e) {
            e.printStackTrace();
        }
        return metricValue;
    }

    private Matcher getMatcher(String openTsdbKey) {
        Matcher m = JMX_ITEM_KEY_PTN.matcher(openTsdbKey);
        if (!m.matches() || m.groupCount() != 2) {
            return null;
        }
        return m;
    }

    private JMXConnector connectWithTimeout(final JMXServiceURL url, long timeout) throws InterruptedException, ExecutionException, TimeoutException {
        ExecutorService executor = Executors.newSingleThreadExecutor();
        Future<JMXConnector> future = executor.submit(new Callable<JMXConnector>() {
            @Override
            public JMXConnector call() throws IOException {
                return JMXConnectorFactory.connect(url);
            }
        });
        return future.get(timeout, TimeUnit.SECONDS);
    }


    public class ItemGetTask implements Callable<JmxCheckDTO> {
        JmxCheckDTO dto;

        public ItemGetTask(JmxCheckDTO dto) {
            this.dto = dto;
            try {
                monitorItemJmxPoll(dto.getHostName(), dto.getHostIp(), dto.getJmxAddr(), dto.getMonitorItem());
            } catch (Exception e) {
                e.printStackTrace();
                LOG.error(e.getMessage(), e);
            }
        }

        @Override
        public JmxCheckDTO call() {
            return dto;
        }
    }
}
